package cn.orangeframe.neon.application.run;

import cn.orangeframe.neon.application.exception.TerminateWorkflowException;
import cn.orangeframe.neon.application.run.async.AsyncPublish;
import cn.orangeframe.neon.application.tasks.WorkflowSystemTask;
import cn.orangeframe.neon.application.tasks.Terminate;
import cn.orangeframe.neon.application.util.IDGenerator;
import cn.orangeframe.neon.application.util.TaskUtils;
import cn.orangeframe.neon.domain.adapter.MetadataAdapter;
import cn.orangeframe.neon.domain.entity.WorkflowDef;
import cn.orangeframe.neon.domain.enums.TaskType;
import cn.orangeframe.neon.domain.enums.WorkflowStatus;
import cn.orangeframe.neon.domain.model.TaskModel;
import cn.orangeframe.neon.domain.model.WorkflowModel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

@Slf4j
@Component
public class WorkflowExecutor {

    private final DeciderService deciderService;

    private final MetadataAdapter metadataAdapter;

    private final AsyncPublish publish;

    public static final String TRACE_KEY = "_traceId";

    public WorkflowExecutor(DeciderService deciderService, MetadataAdapter metadataAdapter, AsyncPublish publish) {
        this.deciderService = deciderService;
        this.metadataAdapter = metadataAdapter;
        this.publish = publish;
    }

    /**
     * 启动工作流
     * @param workflowName 工作流名称
     * @param version 工作流版本
     * @param input 输入参数
     * @return 输出结果
     */
    public Map<String, Object> startWorkflowWithOutput(String workflowName, int version, Map<String, Object> input) {
        WorkflowDef workflowDef = metadataAdapter.getWorkflowDef(workflowName, version);
        return startWorkflowWithOutput(workflowDef, input);
    }

    /**
     * 启动工作流
     * @param workflowDef 工作流定义
     * @param input 输入参数
     * @return 输出结果
     */
    public Map<String, Object> startWorkflowWithOutput(WorkflowDef workflowDef, Map<String, Object> input) {
        String workflowId = IDGenerator.generate();

        Map<String, Object> output = new HashMap<>();
        output.put(TRACE_KEY, workflowId);

        WorkflowModel workflow = new WorkflowModel();
        workflow.setWorkflowName(workflowDef.getName());
        workflow.setWorkflowId(workflowId);
        workflow.setWorkflowDefinition(workflowDef);
        workflow.setStatus(WorkflowStatus.RUNNING);
        workflow.setCreateTime(LocalDateTime.now());
        workflow.setInput(input);
        workflow.setOutput(output);
        try {
            log.debug("start new workflow instance:{} with id:{}", workflow.getWorkflowName(), workflowId);
            decide(workflow);
            log.debug("task size:{}", workflow.getTasks().size());
            log.debug("end workflow instance:{} with id:{}", workflow.getWorkflowName(), workflow.getWorkflowId());
        } catch (Exception e) {
            log.error("execute workflow instance:{}, version:{} error", workflow.getWorkflowName(), workflow.getWorkflowVersion(), e);
        }
        return workflow.getOutput();
    }

    public void decide(WorkflowModel workflow) {
        try {
            DeciderService.DeciderOutcome outcome = deciderService.decide(workflow);
            if (outcome.isCompleted()) {
                endExecution(workflow, outcome.getTerminateTask());
                return;
            }

            List<TaskModel> taskToSchedule = outcome.getTaskToSchedule();

            // 消除重复数据
            taskToSchedule = dedupAndAddTasks(workflow, taskToSchedule);

            if (scheduleTask(workflow, taskToSchedule)) {
                decide(workflow);
            }
        } catch (TerminateWorkflowException e) {
            log.error("decide workflow: {} error", workflow.getWorkflowId(), e);
            terminateWorkflow(workflow, e);
        } catch (Exception e) {
            log.error("decide workflow: {} error", workflow.getWorkflowId(), e);
            throw e;
        }
    }

    /**
     * 调度任务
     * @param workflowModel 流程实例
     * @param taskModels 待调度任务
     * @return 是否调度了任务
     */
    public boolean scheduleTask(WorkflowModel workflowModel, List<TaskModel> taskModels) {
        boolean startedSystemTasks = false;
        if (taskModels == null || taskModels.isEmpty()) {
            return false;
        }

        if (TaskType.TASK_TYPE_FORK.equals(taskModels.getFirst().getTaskType())) {
            // FORK任务异步执行
            CountDownLatch latch = new CountDownLatch(taskModels.size() - 2);
            List<TaskModel> unJoinTask = taskModels.stream()
                    .filter(taskModel -> !TaskType.TASK_TYPE_JOIN.equals(taskModel.getTaskType()))
                    .toList();
            for (TaskModel taskModel : unJoinTask) {
                WorkflowSystemTask workflowSystemTask = WorkflowSystemTask.get(taskModel.getTaskType());
                if (TaskType.TASK_TYPE_FORK.equals(taskModel.getTaskType())) {
                    continue;
                }
                taskModel.setLatch(latch);
                publish.publishAsyncTaskExecute(this, workflowSystemTask, workflowModel, taskModel);
            }
            // 当决策任务为fork_join任务时，返回的任务中，第一个永远是fork任务，最后一个永远是join任务
            TaskModel joinTask = taskModels.getLast();
            joinTask.setLatch(latch);
            WorkflowSystemTask workflowSystemTask = WorkflowSystemTask.get(joinTask.getTaskType());
            workflowSystemTask.execute(workflowModel, joinTask, this);
            startedSystemTasks = true;
        } else {
            for (TaskModel taskModel : taskModels) {
                WorkflowSystemTask workflowSystemTask = WorkflowSystemTask.get(taskModel.getTaskType());
                taskModel.setStartTime(System.currentTimeMillis());
                workflowSystemTask.execute(workflowModel, taskModel, this);
                startedSystemTasks = true;
            }
        }
        return startedSystemTasks;
    }

    /**
     * 结束工作流
     * @param workflow
     * @param terminateTask
     */
    private void endExecution(WorkflowModel workflow, TaskModel terminateTask) {
        if (terminateTask != null) {
            String terminationStatus =
                    (String)
                            terminateTask
                                    .getInputData()
                                    .get(Terminate.getTerminationStatusParameter());
            String reason =
                    (String)
                            terminateTask
                                    .getInputData()
                                    .get(Terminate.getTerminationReasonParameter());
            if (StringUtils.isBlank(reason)) {
                reason =
                        String.format(
                                "Workflow is %s by TERMINATE task: %s",
                                terminationStatus, terminateTask.getTaskId());
            }
            if (WorkflowStatus.FAILED.name().equals(terminationStatus)) {
                workflow.setStatus(WorkflowStatus.FAILED);
            } else {
                workflow.setReasonForInCompletion(reason);
                workflow.setStatus(WorkflowStatus.FAILED);
            }
        } else {
            workflow.setStatus(WorkflowStatus.COMPLETED);
        }
    }

    /**
     * 终止工作流
     * @param workflowModel 流程实例
     * @param e 异常信息
     */
    public void terminateWorkflow(WorkflowModel workflowModel, TerminateWorkflowException e) {
        workflowModel.setStatus(e.workflowStatus);
    }

    /**
     * 消除重复数据
     * @param workflow
     * @param tasks
     * @return
     */
    List<TaskModel> dedupAndAddTasks(WorkflowModel workflow, List<TaskModel> tasks) {
        Set<String> tasksInWorkflow =
                workflow.getTasks().stream()
                        .map(task -> task.getTaskReferenceName() + "_" + task.getRetryCount())
                        .collect(Collectors.toSet());

        // 未完成的do_while任务，需重新调度，但由于workflow中已存在，无需重新放入workflow的task中
        List<TaskModel> doWhileTask = tasks.stream().filter(t ->
                TaskType.TASK_TYPE_DO_WHILE.equals(t.getTaskType())
                && !t.getStatus().isTerminal()).toList();
        List<TaskModel> dedupedTasks =
                tasks.stream()
                        .filter(
                                task ->
                                        !tasksInWorkflow.contains(
                                                task.getTaskReferenceName()
                                                        + "_"
                                                        + task.getRetryCount()))
                        .collect(Collectors.toList());

        workflow.getTasks().addAll(dedupedTasks);
        dedupedTasks.addAll(doWhileTask);
        return dedupedTasks;
    }

    /**
     * 执行循环任务
     * @param loopTask
     * @param workflow
     */
    public void scheduleNextIteration(TaskModel loopTask, WorkflowModel workflow) {
        // Schedule only first loop over task. Rest will be taken care in Decider Service when this
        // task will get completed.
        List<TaskModel> scheduledLoopOverTasks =
                deciderService.getTasksToBeScheduled(
                        workflow,
                        loopTask.getWorkflowTask().getLoopOver().getFirst(),
                        loopTask.getRetryCount(),
                        null);
        scheduledLoopOverTasks.forEach(
                t -> {
                    t.setTaskReferenceName(
                            TaskUtils.appendIteration(
                                    t.getTaskReferenceName(), loopTask.getIteration()));
                    t.setIteration(loopTask.getIteration());
                });
        scheduleTask(workflow, scheduledLoopOverTasks);
        workflow.getTasks().addAll(scheduledLoopOverTasks);
    }
}
